package com.hong.spring.config;

import com.alibaba.fastjson.JSONObject;
import com.hong.spring.listener.UserListener;
import lombok.extern.log4j.Log4j2;
import org.apache.tubemq.client.config.ConsumerConfig;
import org.apache.tubemq.client.consumer.ConsumePosition;
import org.apache.tubemq.client.consumer.ConsumerResult;
import org.apache.tubemq.client.consumer.PullMessageConsumer;
import org.apache.tubemq.client.factory.MessageSessionFactory;
import org.apache.tubemq.client.factory.TubeSingleSessionFactory;
import org.apache.tubemq.corebase.Message;
import org.apache.tubemq.corebase.utils.ThreadUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author: csh
 * @Date: 2021/4/27 18:06
 * @Description:zero配置
 */
@Log4j2
public class TubeMqConsumer {

    //服务地址
    private String addrHost;
    //topic
    private String topic;
    //消费组
    private String group;
    //
    private MessageSessionFactory messageSessionFactory;
    //
    PullMessageConsumer messagePullConsumer;

    @Autowired
    private UserListener userListener;


    public TubeMqConsumer(String addrHost, String topic, String group) {
        this.addrHost = addrHost;
        this.topic = topic;
        this.group = group;
    }

    public void init(){
       try {
           final ConsumerConfig consumerConfig = new ConsumerConfig(addrHost, group);
           consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
           messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
           messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
           messagePullConsumer.subscribe(topic, null);
           messagePullConsumer.completeSubscribe();
           recvStr();
       }catch (Exception e){
            log.error("tubemq初始化失败",e);
       }
    }

    @Async
    public void recvStr(){
        try {
            log.info("开始自动拉取消息！");

            while (true){
                ConsumerResult result = messagePullConsumer.getMessage();
                if (result.isSuccess()) {
                    List<Message> messageList = result.getMessageList();
                    for (Message message : messageList) {
                        log.info("接收到的消息: " + JSONObject.toJSONString(message)+"内容是:"+new String(message.getData()));
                        userListener.Listener(new String(message.getData()));
                    }
                    messagePullConsumer.confirmConsume(result.getConfirmContext(), true);
                } else {
                    if (result.getErrCode() == 400) {
                        ThreadUtils.sleep(100);
                    } else {
                        if (result.getErrCode() != 404) {
                            log.error("Receive messages errorCode is %d, Error message is %s", result.getErrCode(), result.getErrMsg());
                        }
                    }
                }
            }
        }catch (Exception e){
            log.error("接收消失败请重试{}",e);
        }

    }
    /**
     *
     * 功能描述: 关闭接口
     *
     * @param:
     * @return:
     * @auther: csh
     * @date: 2021/5/18 14:25
     */
    public void close(){
        try {
            if(messageSessionFactory!=null){
                messageSessionFactory.shutdown();
            }
        }catch (Exception e){
            log.error("关闭tubemq失败{}",e);
        }
    }

}
